from celery import shared_task
from celery.signals import worker_ready, task_prerun, task_postrun, task_failure
from datetime import datetime, timedelta
import logging
import traceback
import sys
import os
from .constant import *

logger = logging.getLogger(__name__)

# 添加任务预运行信号处理
@task_prerun.connect
def task_prerun_handler(task_id, task, *args, **kwargs):
    print(f"\n[Celery信号] 任务准备运行: {task.name} (ID: {task_id})")
    logger.info(f"任务准备运行: {task.name} (ID: {task_id})")

# 添加任务后运行信号处理
@task_postrun.connect
def task_postrun_handler(task_id, task, *args, retval=None, state=None, **kwargs):
    print(f"[Celery信号] 任务运行完成: {task.name} (ID: {task_id}), 状态: {state}")
    logger.info(f"任务运行完成: {task.name} (ID: {task_id}), 状态: {state}")
    
    # 如果是市场数据分析任务完成，发送WebSocket通知
    if task.name == 'data_analysis.tasks.analyze_market_data':
        from channels.layers import get_channel_layer
        from asgiref.sync import async_to_sync
        
        try:
            channel_layer = get_channel_layer()
            # 同时使用直接发送和组播两种方式
            async_to_sync(channel_layer.send)(
                'market_data_updates',  # 使用相同的组名
                {
                    'type': 'send.progress',  # 使用点号分隔的命名方式
                    'message': '市场数据分析任务已完成',
                    'task_id': task_id,
                    'status': state,
                    'should_refresh': True
                }
            )
            
            async_to_sync(channel_layer.group_send)(
                'market_data_updates',
                {
                    'type': 'task_completed',
                    'message': '市场数据分析任务已完成',
                    'task_id': task_id,
                    'status': state,
                    'should_refresh': True
                }
            )
            logger.info(f"已发送WebSocket通知: 市场数据分析任务完成 (ID: {task_id} ,time {datetime.now()})")
        except Exception as e:
            logger.error(f"发送WebSocket通知失败: {str(e)}")

# 添加任务失败信号处理
@task_failure.connect
def task_failure_handler(task_id, exception, traceback, einfo, *args, **kwargs):
    print(f"\n[Celery信号] 任务失败: (ID: {task_id})")
    print(f"异常: {exception}")
    print(f"错误信息: {einfo}")
    logger.error(f"任务失败: (ID: {task_id}), 异常: {exception}, 错误信息: {einfo}")


# 一旦celery启动，就开始执行该函数。用于后台服务启动后自动进行数据分析，存储到数据库中
# 然后，在通过前端访问，将分析好的数据展示在前端页面中
@worker_ready.connect
def at_start(sender, **kwargs):
    """
    当Celery worker启动时自动执行每日数据分析任务
    """
    print(f"\n{'*'*50}")
    print(f"Celery worker启动事件触发 - {datetime.now()}")
    print(f"sender: {sender}")
    print(f"kwargs: {kwargs}")
    
    try:
        logger.info("Celery worker已启动，准备执行每日数据分析任务...")
        print("准备调用统一入口函数 start_analysis_tasks...")
        
        # 调用统一入口函数
        start_analysis_tasks()

    except Exception as e:
        error_msg = f"提交任务时出错: {str(e)}"
        print(f"\n{'!'*50}")
        print(error_msg)
        print(f"{'!'*50}\n")
        logger.error(f"{error_msg}\n{traceback.format_exc()}")
    
    print(f"{'*'*50}\n")

def start_analysis_tasks():
    """
    统一入口函数，用于启动所有数据分析任务
    """
    try:
        # 1.分析市场每日数据。延迟30秒执行初始任务，确保前端有足够时间建立WebSocket连接
        task = analyze_market_data.apply_async(
            args=['day'],
            countdown=2,  # 延迟2秒执行
            expires=600,  # 10分钟后过期
            retry=True,
            retry_policy={
                'max_retries': 3,
                'interval_start': 0,
                'interval_step': 0.2,
                'interval_max': 0.5,
            }
        )
        print(f"任务已提交，任务ID: {task.id}")
        print(f"任务状态: {task.status}")
        logger.info(f"每日数据分析任务已提交，任务ID: {task.id}")
        
        # 2.分析市场量能数据
        analyze_market_energy.apply_async(args=['day'])
        
        # 3.未来可以在此添加其他任务
        
    except Exception as e:
        error_msg = f"提交任务时出错: {str(e)}"
        print(f"\n{'!'*50}")
        print(error_msg)
        print(f"{'!'*50}\n")
        logger.error(f"{error_msg}\n{traceback.format_exc()}")
        raise


@shared_task(
    bind=True, 
    max_retries=3, 
    default_retry_delay=5,
    autoretry_for=(Exception,),
    retry_backoff=True,
    retry_backoff_max=60,
    retry_jitter=True,
    # Windows平台不支持soft_time_limit，移除此配置
    time_limit=900,       # 15分钟硬时间限制
    acks_late=True,       # 任务完成后再确认
    track_started=True    # 跟踪任务开始状态
)
def analyze_market_energy(self, period='day'):
    """
    执行市场能量数据分析任务，统计沪深、上证、深证、创业板四个指数的量能数据
    
    Args:
        self: Celery任务实例
        period (str): 分析周期，可选值：'day', 'week', 'month'
    """
    import traceback,sys,os
    from django.db.models import Avg, Max, Min, Count, Sum, F, Q
    from tushare_data.models import IndexDaily, TradeDate
    from data_analysis.models import DataAnalysisRecord, AmountEnergy

    print(f"开始执行 {period} 市场能量数据分析任务 - {datetime.now()}")
    print(f"任务ID: {self.request.id}")
    print(f"任务名称: {self.name}")
    print(f"当前工作目录: {os.getcwd()}")
    print(f"环境变量: DJANGO_SETTINGS_MODULE={os.environ.get('DJANGO_SETTINGS_MODULE', '未设置')}")
    
    logger.info(f"开始执行市场能量数据分析任务，周期: {period}, 任务ID: {self.request.id}")
    
    try:
        logger.info(f"开始执行{period}市场能量数据分析任务 - {datetime.now()}")
        
        # 获取最近的交易日期
        logger.info("正在获取最近的交易日期...")
        latest_index = IndexDaily.objects.order_by('-trade_date').first()
        if not latest_index:
            error_msg = "错误: 没有找到指数数据"
            logger.error(error_msg)
            return False
            
        analysis_date = latest_index.trade_date
        logger.info(f"获取到最新交易日期: {analysis_date}, 分析周期: {period}")

        # 获取最新分析记录
        latest_analysis = DataAnalysisRecord.objects.filter(
            data_name=DATA_CLASS_TYPES['MARKET_ENERGY'],
        ).order_by('-last_trade_date').first()
        
        if latest_analysis:
            start_date = latest_analysis.last_trade_date.strftime('%Y%m%d')
        else:
            start_daily = IndexDaily.objects.order_by('-trade_date').last()#第一次的时候取最开始的交易日，即20200103
            start_date = start_daily.trade_date
            logger.info(f"获取到起始交易日期: {start_date}")
        print(f'开始日期: {start_date}')
        
        # 获取需要分析的交易日范围
        analysis_dates = TradeDate.objects.filter(
            is_open=True,
            cal_date__gte=start_date,
            cal_date__lte=analysis_date
        ).order_by('cal_date').values_list('cal_date', flat=True)
        
        logger.info(f"需要分析 {len(analysis_dates)} 个交易日的市场能量数据")
        
        # 对每个交易日进行分析
        for trade_date in analysis_dates:
            logger.info(f'开始分析 {trade_date} 市场能量数据')
            try:
                # 获取当日指数数据
                index_data = IndexDaily.objects.filter(trade_date=trade_date)
                
                # 统计沪深、上证、深证、创业板的量能数据
                sh_amount = index_data.filter(ts_code='000001.SH').values_list('amount', flat=True).first() or 0
                sz_amount = index_data.filter(ts_code='399001.SZ').values_list('amount', flat=True).first() or 0
                cyb_amount = index_data.filter(ts_code='399006.SZ').values_list('amount', flat=True).first() or 0
                hs_amount = sh_amount + sz_amount
                # 保存量能数据
                AmountEnergy.objects.update_or_create(
                    date=datetime.strptime(trade_date,'%Y%m%d').date(),
                    defaults={
                        'hs_amount': hs_amount,
                        'sh_amount': sh_amount,
                        'sz_amount': sz_amount,
                        'cyb_amount': cyb_amount,
                    }
                )
                logger.info(f"成功保存 {trade_date} 市场能量数据")
            except Exception as e:
                error_msg = f"分析 {trade_date} 市场能量数据时出错: {str(e)}"
                logger.error(f"{error_msg}\n{traceback.format_exc()}")
                continue
        
        # 更新分析记录.由于tushare的数据都是字符串YYMMdd格式，现在分析时用的都是date格式。所以需要从字符串到date的转换。
        DataAnalysisRecord.objects.update_or_create(
            data_name=DATA_CLASS_TYPES['MARKET_ENERGY'],
            data_type=period,
            defaults={
                'last_trade_date': datetime.strptime(analysis_date,'%Y%m%d').date(),
            }
        )
        logger.info(f"成功更新市场能量数据分析记录，最新交易日期: {analysis_date}")
        
        return True
    except Exception as e:
        error_msg = f"执行市场能量数据分析任务时出错: {str(e)}"
        print(f"\n{'!'*50}")
        print(error_msg)
        print(f"{'!'*50}\n")
        logger.error(f"{error_msg}\n{traceback.format_exc()}")
        return False

@shared_task(
    bind=True, 
    max_retries=3, 
    default_retry_delay=5,
    autoretry_for=(Exception,),
    retry_backoff=True,
    retry_backoff_max=60,
    retry_jitter=True,
    # Windows平台不支持soft_time_limit，移除此配置
    time_limit=900,       # 15分钟硬时间限制
    acks_late=True,       # 任务完成后再确认
    track_started=True    # 跟踪任务开始状态
)
def analyze_market_data(self, period='day'):
    """
    执行市场数据分析任务
    
    Args:
        self: Celery任务实例
        period (str): 分析周期，可选值：'day', 'week', 'month'
    """
    import traceback,sys,os
    from django.db.models import Avg, Max, Min, Count, Sum, F, Q
    from tushare_data.models import DailyData, LimitList, IndexDaily,TradeDate
    from data_analysis.models import MarketSummary, DataAnalysisRecord

    print(f"开始执行 {period} 数据分析任务 - {datetime.now()}")
    print(f"任务ID: {self.request.id}")
    print(f"任务名称: {self.name}")
    print(f"当前工作目录: {os.getcwd()}")
    print(f"环境变量: DJANGO_SETTINGS_MODULE={os.environ.get('DJANGO_SETTINGS_MODULE', '未设置')}")
    
    logger.info(f"开始执行数据分析任务，周期: {period}, 任务ID: {self.request.id}")
    
    try:
        logger.info(f"开始执行{period}数据分析任务 - {datetime.now()}")
        
        # 获取最近的交易日期
        logger.info("正在获取最近的交易日期...")
        latest_daily = DailyData.objects.order_by('-trade_date').first()
        if not latest_daily:
            error_msg = "错误: 没有找到交易数据"
            logger.error(error_msg)
            return False
            
        analysis_date = latest_daily.trade_date
        logger.info(f"获取到最新交易日期: {analysis_date}, 分析周期: {period}")

        # 获取最新分析记录
        latest_analysis = DataAnalysisRecord.objects.filter(
            data_name=DATA_CLASS_TYPES['MARKET_SUMMARY'],
        ).order_by('-last_trade_date').first()
        
        if latest_analysis:
            start_date = latest_analysis.last_trade_date.strftime('%Y%m%d')
        else:
            start_daily = DailyData.objects.order_by('-trade_date').last()#第一次的时候取最开始的交易日
            start_date = start_daily.trade_date

            logger.info(f"获取到起始交易日期: {start_date}")
        print(f'开始日期: {start_date}')
        
        # 获取需要分析的交易日范围
        analysis_dates = TradeDate.objects.filter(
            is_open=True,
            cal_date__gte=start_date,
            cal_date__lte=analysis_date
        ).order_by('cal_date').values_list('cal_date', flat=True)
        
        logger.info(f"需要分析 {len(analysis_dates)} 个交易日的市场数据")
        
        # 对每个交易日进行分析
        for trade_date in analysis_dates:
            logger.info(f'开始分析 {trade_date} 市场数据')
            try:
                # 获取当日市场数据
                daily_data = DailyData.objects.filter(trade_date=trade_date)
                index_data = IndexDaily.objects.filter(trade_date=trade_date)
                limit_data = LimitList.objects.filter(trade_date=trade_date)
                # 获取涨跌停数据
                limit_up = limit_data.filter(limit='U').count()
                limit_down = limit_data.filter(limit='D').count()

                #开始按照顺序进行统计，首先获取指数涨跌幅
                sh_index = index_data.filter(ts_code='000001.SH').first()
                sz_index = index_data.filter(ts_code='399001.SZ').first()
                cyb_index = index_data.filter(ts_code='399006.SZ').first()
                # 计算市场指标
                total_stocks = daily_data.count()
                
                # 详细涨跌幅区间统计
                up_7_pct = daily_data.filter(pct_chg__gt=7).count()
                up_5_to_up_7_pct = daily_data.filter(pct_chg__gt=5, pct_chg__lte=7).count()
                up_2_to_5_pct = daily_data.filter(pct_chg__gt=2, pct_chg__lte=5).count()
                up_0_to_2_pct = daily_data.filter(pct_chg__gt=0, pct_chg__lte=2).count()
                flat_stocks = daily_data.filter(pct_chg=0).count()
                down_0_to_2_pct = daily_data.filter(pct_chg__lt=0, pct_chg__gte=-2).count()
                down_2_to_5_pct = daily_data.filter(pct_chg__lt=-2, pct_chg__gte=-5).count()
                down_5_to_7_pct = daily_data.filter(pct_chg__lt=-5, pct_chg__gte=-7).count()
                down_7_pct = daily_data.filter(pct_chg__lt=-7).count()
                
                up_stocks = daily_data.filter(pct_chg__gt=0).count()
                down_stocks = daily_data.filter(pct_chg__lt=0).count()
                sh_index_amount = sh_index.amount if sh_index else 0
                sz_index_amount = sz_index.amount if sz_index else 0
                
                # 创建或更新市场分析记录
                trade_date_dt = datetime.strptime(trade_date, '%Y%m%d')
                MarketSummary.objects.update_or_create(
                    date=trade_date_dt,
                    defaults={
                        'sh_index': sh_index.close if sh_index else 0,
                        'sh_change_pct': sh_index.pct_chg if sh_index else 0,
                        'sz_index': sz_index.close if sz_index else 0,
                        'sz_change_pct': sz_index.pct_chg if sz_index else 0,
                        'cyb_index': cyb_index.close if cyb_index else 0,
                        'cyb_change_pct': cyb_index.pct_chg if cyb_index else 0,
                        'total_stocks': total_stocks,
                        'up_stocks': up_stocks,
                        'down_stocks': down_stocks,
                        'limit_up_stocks': limit_up,
                        'limit_down_stocks': limit_down,
                        'up_7_pct': up_7_pct,
                        'up_5_to_up_7_pct': up_5_to_up_7_pct,
                        'up_2_to_5_pct': up_2_to_5_pct,
                        'up_0_to_2_pct': up_0_to_2_pct,
                        'flat_stocks': flat_stocks,
                        'down_0_to_2_pct': down_0_to_2_pct,
                        'down_2_to_5_pct': down_2_to_5_pct,
                        'down_5_to_7_pct': down_5_to_7_pct,
                        'down_7_pct': down_7_pct,
                        'total_amount': sh_index_amount + sz_index_amount,
                    }
                )
                
            except Exception as e:
                logger.error(f"分析交易日 {trade_date} 时出错: {str(e)}")
                continue
                
        # 更新分析记录
        DataAnalysisRecord.update_record(
            data_name=DATA_CLASS_TYPES['MARKET_SUMMARY'],
            data_type='day',  # 使用预定义的日度分析类型
            last_trade_date= datetime.strptime(analysis_date, '%Y%m%d')
        )
        logger.info(f"数据分析任务执行完成 - {datetime.now()}")
    except Exception as e:
        error_msg = f"数据分析任务执行失败: {str(e)}"
        logger.error(error_msg)
        logger.error(traceback.format_exc())
        # 重新抛出异常，让Celery处理重试
        raise
 